package com.storm.queue

import com.fasterxml.jackson.databind.ObjectMapper
import com.storm.queue.RedisQueueKeys.RD_LIST_TOPIC_PRE_LOCK
import com.storm.queue.core.DelayOperation
import com.storm.queue.model.Job
import org.slf4j.LoggerFactory
import org.springframework.data.redis.core.RedisTemplate
import org.springframework.util.StringUtils
import java.util.*
import java.util.stream.Collectors

class RedisDelayOperation(private val redisTemplate: RedisTemplate<String, Any>,
                          private val redisKeyUtil: RedisKeyUtil,
                          private val objectMapper: ObjectMapper,
                          private val redisLock: RedisLock):DelayOperation {
    val log = LoggerFactory.getLogger(this.javaClass)
    override fun getJob(jobId: String): Job? {
        val job = redisTemplate.opsForHash<String, Any>()[redisKeyUtil.getDelayQueueTableKey(), jobId] ?: return null
        return objectMapper.convertValue(job, Job::class.java)
    }

    override fun queuePop(topic: String): String? {
        val topicListKey = redisKeyUtil.getTopicListKey(topic)
        val jobIdResult = redisTemplate.opsForList().rightPop(topicListKey) ?: return null
        return jobIdResult.toString()
    }

    override fun retryJob(topic: String, job: Job) {
        val topicId: String = redisKeyUtil.getTopicId(topic, job.id)
        //Job池放入一个Job
        //Job池放入一个Job
        redisTemplate.opsForHash<Any, Any>().put(redisKeyUtil.getDelayQueueTableKey(), topicId, job)
        redisTemplate.opsForList().leftPush(redisKeyUtil.getTopicListKey(topic), topicId)
        log.info("新增重试Job成功:topicId =>{}", topicId)
    }

    override fun setJobLast() {
        val lockValue: String = redisLock.getLockValue()
        try {
            if (redisLock.lock(RD_LIST_TOPIC_PRE_LOCK, lockValue)) {
                redisTemplate.opsForHash<String, Any>().entries(redisKeyUtil.getDelayQueueTableKey())
                        .forEach { (k: Any, _: Any) ->
                            val score = redisTemplate.opsForZSet().score(redisKeyUtil.getBucketKey(), k)
                            if (score == null) {
                                redisTemplate.opsForZSet().add(redisKeyUtil.getBucketKey(), k, System.currentTimeMillis().toDouble())
                            }
                        }
            }
        } catch (e: Exception) {
            log.error(e.message, e)
        } finally {
            redisLock.unlock(RD_LIST_TOPIC_PRE_LOCK, lockValue)
        }
    }

    override fun deleteJob(topic: String, id: String) {
        if (StringUtils.isEmpty(id) || StringUtils.isEmpty(topic)) {
            return
        }
        val topicId: String = redisKeyUtil.getTopicId(topic, id)
        //Job池删除置顶Job
        redisTemplate.opsForHash<Any, Any>().delete(redisKeyUtil.getDelayQueueTableKey(), topicId)
        //Bucket放入Job的时间戳
        redisTemplate.opsForZSet().remove(redisKeyUtil.getBucketKey(), topicId)
        log.info("删除成功 Job:topicId =>{}", topicId)
    }

    override fun moveToQueue(): Long {
        val lockValue: String = redisLock.getLockValue()
        return try {
            if (redisLock.lock(RD_LIST_TOPIC_PRE_LOCK, lockValue)) {
                val readyList: List<String> = getReadyList()
                if (readyList.isEmpty()) {
                    return Long.MAX_VALUE
                }
                readyList.forEach{ member: String ->
                    log.info("处理一个新元素：{}", member)
                    redisTemplate.opsForList().leftPush(redisKeyUtil.getTopicListKeyByMember(member), member!!)
                    redisTemplate.opsForZSet().remove(redisKeyUtil.getBucketKey(), member)
                    log.info("List队列{}push一个新元素{}", redisKeyUtil.getTopicListKeyByMember(member), member)
                }
                //最后查一次返回
                val lastOnes = redisTemplate.opsForZSet().range(redisKeyUtil.getBucketKey(), 0L, 1L)
                if (lastOnes == null || lastOnes.isEmpty()) {
                    return Long.MAX_VALUE
                }
                val it2: Iterator<Any> = lastOnes.iterator()
                val m2 = it2.next()
                val score2 = redisTemplate.opsForZSet().score(redisKeyUtil.getBucketKey(), m2) ?: return Long.MAX_VALUE
                score2.toLong()
            } else {
                0L
            }
        } catch (e: Exception) {
            Long.MAX_VALUE
        } finally {
            redisLock.unlock(RD_LIST_TOPIC_PRE_LOCK, lockValue)
        }
    }

    /**
     * 获取可以操作的列表 一次只操作1000个
     * @return 操作列表 notnull
     */
    private fun getReadyList(): List<String> {
        try {
            val objects = redisTemplate.opsForZSet().rangeByScore(redisKeyUtil.getBucketKey(), 0.0, System.currentTimeMillis().toDouble(), 0, 200)
            if (objects != null) {
                val strings = objects
                        .stream()
                        .map { it.toString() }
                        .collect(Collectors.toList())
                log.info("读取到的topic is {}", strings)
                return strings
            }
        } catch (e: Exception) {
            log.error(e.message, e)
        }
        return ArrayList()
    }


    override fun addJob(job: Job, topic: String, limitTime: Long) {
        if (StringUtils.isEmpty(job.id) || StringUtils.isEmpty(topic)) {
            return
        }
        val topicId: String = redisKeyUtil.getTopicId(topic, job.id)
        //Job池放入一个Job
        //Job池放入一个Job
        redisTemplate.opsForHash<Any, Any>().put(redisKeyUtil.getDelayQueueTableKey(), topicId, job)
        //Bucket放入Job的时间戳
        //Bucket放入Job的时间戳
        val add = redisTemplate.opsForZSet().add(redisKeyUtil.getBucketKey(), topicId, limitTime.toDouble()) ?: false
        if (add) {
            log.info("新增成功Job:topicId =>{}", topicId)
        } else {
            log.info("桶数据添加失败")
        }
    }
}